package com.wdl.networkflow.uv

import com.wdl.networkflow.pv.UserBehavior
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import java.lang

/**
 * 定义全窗口函数
 */
class UvCountResult() extends AllWindowFunction[UserBehavior, UvCount, TimeWindow] {
  override def apply(window: TimeWindow,
                     input: Iterable[UserBehavior],
                     out: Collector[UvCount]): Unit = {
    /** 定义一个 set 类型来保存所有的 userId，自动去重 */
    var userIds: Set[Long] = Set[Long]()
    /** 将当前窗口的所有数据，添加到 set 里 */
    for (elem <- input) {
      userIds += elem.userId
    }

    /** 输出 set 的大小，就是去重之后的 uv 值 */
    out.collect(UvCount(window.getEnd, userIds.size))
  }
}
